package com.example.mq.mqServer;

import com.example.mq.common.Consumer;
import com.example.mq.mqServer.center.ConsumerCenter;
import com.example.mq.mqServer.core.Binding;
import com.example.mq.mqServer.core.MSGQueue;
import com.example.mq.mqServer.core.exchange.Exchange;
import com.example.mq.mqServer.core.exchange.ExchangeType;
import com.example.mq.mqServer.core.message.BasicProperties;
import com.example.mq.mqServer.core.message.Message;
import com.example.mq.mqServer.center.DiskDataCenter;
import com.example.mq.mqServer.center.MemoryDataCenter;
import com.example.mq.mqServer.exception.MqException;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;


import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @BelongsProject: 0.mq
 * @BelongsPackage: com.example.mq.mqServer
 * @Author : 游在海里的鱼
 * @CreateTime : 2023-12-16  19:04
 * @Version : 1.0
 * @ClassName : 服务端虚拟机类
 * @Description : 消息队列的虚拟服务端
 */
@Slf4j
@Data
public class VirtualHost {
    //虚拟机名称，也是虚拟机的唯一标识符
    private String virtualHostName;
    //内存数据中心
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    //硬盘数据中心
    private DiskDataCenter diskDataCenter= new DiskDataCenter();
    //消费者管理中心
    private ConsumerCenter consumerCenter = new ConsumerCenter(this);
    //路由器，控制消息塞进队列的路由
    private Router router = new Router();

    //交换机锁和队列锁
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();


/** ---------------------------------------------------------------------------------------------------------------------
 * @Method  : 构造方法
 * @Describe: 虚拟机构造
 * @modifier: [public] [构造方法]
**/
public VirtualHost(String virtualHostName) {
        //1.虚拟机命名
        this.virtualHostName = virtualHostName;
        //2.硬盘和内存数据中心初始化
        diskDataCenter.init();
        memoryDataCenter.init(diskDataCenter);
        log.info("[init] [虚拟机:"+virtualHostName+"]初始化完成");

    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 增加虚拟机前缀
     * @Describe: 在输入的字符串前拼接上虚拟机的唯一ID
     * @modifier: [private] [实例方法]
    **/private String addPrefix(String name){
        return virtualHostName+name;
    }
    
    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 获取此虚拟机的内存数据中心
     * @Describe: 消费者中心需要用到内存
     * @modifier: [public] [实例方法]
    **/
    public MemoryDataCenter getMemoryDataCenter(){
        return memoryDataCenter;
    }
    
    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 声名交换机（创建交换机）
     * @Describe: 如果交换机已存在,声明并不需要报异常,而创建需要报异常
     * @modifier: [public] [实例方法]
    **/
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String,Object> arguments){
        //1.添加虚拟机前缀
        exchangeName = addPrefix(exchangeName);
        //2.交换机锁
        synchronized (exchangeLocker){
            //3.判断交换机是否存在,存在则结束
            if (memoryDataCenter.getExchange(exchangeName) != null) {
                log.info("[exchangeDeclare] [交换机:"+exchangeName+"]已存在");
                return true;
            }
            //4.创建交换机对象
            Exchange exchange = new Exchange(exchangeName, exchangeType, durable, autoDelete, arguments);
            //5.持久化
            //先数据库后内存，以防内存加载成功而持久化失败，导致要去删除数据库数据
            if (durable) {
                diskDataCenter.insertExchange(exchange);
            }
            //6.加载到内存
            memoryDataCenter.addExchange(exchange);
            log.info("[exchangeDeclare] [类型:"+exchangeType.name()+"][名称:"+exchangeName+"]的交换机创建成功");
            return true;
        }

    }
    
    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 删除交换机
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean exchangeDelete(String exchangeName){
        exchangeName = addPrefix(exchangeName);
        synchronized (exchangeLocker) {
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                new MqException("[VirtualHost] [exchangeDelete] " +
                        "[交换机:"+exchangeName+"]删除失败,因为此交换机不存在").printStackTrace();
            }
            if(exchange.isDurable()){
                diskDataCenter.deleteExchange(exchangeName);
            }
            memoryDataCenter.removeExchange(exchangeName);
            log.info("[exchangeDelete] [交换机:"+exchangeName+"]删除成功");
            return true;
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 声明队列
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String,Object> arguments){
        //1.添加虚拟机前缀
        queueName = addPrefix(queueName);
        //2.加锁
        synchronized (queueLocker) {
            //3.判断队列是否存在
            if(memoryDataCenter.getQueue(queueName) != null){
                log.info("[queueDeclare] [队列:"+queueName+"]已存在");
                return true;
            }
            //4.创建队列对象
            MSGQueue queue = new MSGQueue(queueName,durable,exclusive,autoDelete,arguments);
            //5.持久化
            //先数据库后内存，以防内存加载成功而持久化失败，导致要去删除数据库数据
            if(durable){
                diskDataCenter.insertQueue(queue);
            }
            //6.加载到内存
            memoryDataCenter.addQueue(queue);
            log.info("[queueDeclare] [队列:"+queueName+"]创建成功");
            return true;
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 删除队列
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean queueDelete(String queueName){
        queueName = addPrefix(queueName);
        synchronized (queueLocker) {
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                new MqException("[VirtualHost] [queueDelete] " +
                        "[队列:"+queueName+"]删除失败,因为此队列不存在").printStackTrace();
                return false;
            }
            if(queue.isDurable()){
                diskDataCenter.deleteQueue(queueName);
            }
            memoryDataCenter.removeQueue(queueName);
            log.info("[queueDelete] [队列:"+queueName+"]删除成功");
            return true;
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 添加绑定
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean queueBind(String exchangeName, String queueName,String bindingKey){
        //1.添加虚拟机前缀
        exchangeName = addPrefix(exchangeName);
        queueName = addPrefix(queueName);
        //2.加锁
        synchronized (exchangeLocker) {
            synchronized (queueLocker) {
                //3.判断绑定是否存在
                if(memoryDataCenter.getBinding(exchangeName,queueName) != null){
                    new MqException("[VirtualHost] [queueBind] [队列:"+queueName+"]和" +
                            "[交换机:"+exchangeName+"]绑定失败,因为此绑定已存在").printStackTrace();
                    return false;
                }
                if(!router.checkBindingKey(bindingKey)){
                    new MqException("[VirtualHost] [queueBind] [队列:"+queueName+"]和" +
                            "[交换机:"+exchangeName+"]绑定失败,因为[bindingKey"+bindingKey+"]不合法").printStackTrace();
                    return false;
                }

                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if( exchange == null){
                    new MqException("[VirtualHost] [queueBind] [队列:"+queueName+"]和" +
                            "[交换机:"+exchangeName+"]绑定失败,因为[交换机:]不存在").printStackTrace();
                    return false;
                }
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    new MqException("[VirtualHost] [queueBind] [队列:"+queueName+"]和" +
                            "[交换机:"+exchangeName+"]绑定失败,因为[队列:]不存在").printStackTrace();
                    return false;
                }

                //4.创建绑定对象
                Binding binding = new Binding(exchangeName,queueName,bindingKey);
                //5.持久化
                if(exchange.isDurable() && queue.isDurable()){
                    diskDataCenter.insertBinding(binding);
                }
                //6.加载到内存
                memoryDataCenter.addBinding(binding);
                log.info("[queueBind] [队列:"+queueName+"]绑定[交换机:"+exchangeName+"]，[bindingKey:"+bindingKey+"]");
                return true;
            }
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 取消绑定
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean queueUnBind(String exchangeName,String queueName){
        exchangeName = addPrefix(exchangeName);
        queueName = addPrefix(queueName);
        synchronized (exchangeLocker) {
            synchronized (queueLocker) {
                Binding binding = memoryDataCenter.getBinding(exchangeName,queueName);
                if(binding == null){
                    new MqException("[VirtualHost] [queueUnBind] [队列:"+queueName+"]和" +
                            "[交换机:"+exchangeName+"]解绑失败,因为此绑定不存在").printStackTrace();
                    return false;
                }
                if(diskDataCenter.selectOneBinding(binding)!=null){
                    diskDataCenter.deleteBinding(binding);
                }
                memoryDataCenter.removeBinding(binding);
                log.info("[queueUnBind] [队列:"+queueName+"]和[交换机:"+exchangeName+"]解绑");
                return true;
            }
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 发送消息
     * @Describe: routingKey：直接交换机是队列名，扇形交换机直接忽略，主题交换机就算没有通配符的路由
     * @modifier: [public] [实例方法]
    **/
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties,byte[] data){
        //1.交换机名称增加前缀
        exchangeName = addPrefix(exchangeName);
        //2.判断routingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                new MqException("[VirtualHost] [basicPublish]往" + "[交换机:"+exchangeName+"]" +
                        "发送消息失败,因为[消息routingKey: "+routingKey+"]不合法").printStackTrace();
                return false;
            }
        //3.判断交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                new MqException("[VirtualHost] [basicPublish] 往" +
                        "[交换机:"+exchangeName+"]发送消息失败,因为此交换机不存在").printStackTrace();
                return false;
            }
        //4.发送消息
            //4.1 直接交换机
            if(exchange.getType() == ExchangeType.DIRECT){
                //获取队列对象
                String queueName = addPrefix(routingKey);
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    new MqException("[VirtualHost] [basicPublish] 往" +
                            "[队列:"+queueName+"]发送消息失败,因为此队列不存在").printStackTrace();
                    return false;
                }
                //构建消息对象
                Message message = Message.MessageFactory(routingKey,basicProperties,data);
                //发送消息到队列
                sendMessage(queue,message);
                log.info("[basicPublish] [消息:"+message.getId()+"][routingKey:"+message.getRoutingKey()+"]" +
                        "往[交换机:"+exchangeName+"][队列:"+queueName+"]发送成功,");
            //4.2 fanout或topic交换机
            }else{
                //获得此交换机下的所有绑定
                ConcurrentHashMap<String,Binding>queue_binding = memoryDataCenter.getAllBinding(exchangeName);
                //遍历所有绑定
                for(Binding binding:queue_binding.values()){
                    //获得队列对象,根据绑定
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null){
                        log.error("[basicPublish] 往[队列::"+binding.getQueueName()+"]发送消息失败,因为此队列不存在");
                        continue;
                    }
                    //构建消息
                    Message message = Message.MessageFactory(routingKey,basicProperties,data);
                    //路由匹配
                    if(!router.router(exchange.getType(),binding,message)){
                        log.error("[basicPublish] 往[队列::"+binding.getQueueName()+"]发送消息失败,因为[路由router]不匹配");
                        continue;
                    }
                    //发送消息
                    sendMessage(queue,message);
                    log.info("[basicPublish] [消息:"+message.getId()+"][routingKey:"+message.getRoutingKey()+"]" +
                            "往[交换机:"+exchangeName+"][队列:"+queue.getName()+"]发送成功,");
                }
            }
            return true;
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 消息加载到内存和磁盘中
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private boolean sendMessage(MSGQueue queue , Message message){
        if(message.getDeliverMode() == 2){
            diskDataCenter.insertMessage(queue,message);
        }
        //将消息添加进内存
        memoryDataCenter.offerMessage(queue,message);
        //通知消费者中心，推送消息给此队列的消费者
        consumerCenter.notifyConsume(queue.getName());
        return true;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 订阅消息
     * @Describe: consumerTag == channel的唯一标识符
     * @modifier: [public] [实例方法]
    **/
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        queueName = addPrefix(queueName);
        MSGQueue queue = memoryDataCenter.getQueue(queueName);
        if(queue == null){
            new MqException("[VirtualHost] [basicConsume] [队列:"+queueName+"]不存在").printStackTrace();
            return false;
        }
        consumerCenter.addConsumer(consumerTag,queue,autoAck,consumer);
        log.info("[basicConsume] [消费者"+consumerTag+"]订阅[队列:"+queueName+"]");
        return true;
    }


    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 消息应答
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public boolean basicAck(String queueName,String messageId){
        queueName = addPrefix(queueName);
        MSGQueue queue = memoryDataCenter.getQueue(queueName);
        if (queue == null){
            new MqException("[VirtualHost] [basicAck] [队列:"+queueName+"]不存在").printStackTrace();
        }
        Message message = memoryDataCenter.getUnACK(queueName,messageId);
        if (message == null){
            new MqException("[VirtualHost] [basicAck] " +
                    "未应答哈希表中，没有[队列:"+queueName+"]的[消息"+messageId+"]").printStackTrace();
        }
        if(message.getDeliverMode() == 2){
            diskDataCenter.deleteMessage(queue,message);
        }
        memoryDataCenter.removeMessage(messageId);
        memoryDataCenter.removeUnACK(queueName,messageId);
        log.info("[basicAck] [队列:"+queue.getName()+"]的[消息"+message.getId()+"]消费后成功应答");
        return true;
    }
}
